package com.twoandone.tms.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.twoandone.tms.realtime.app.func.DimAsyncFunction;
import com.twoandone.tms.realtime.beans.OrderCityBean;
import com.twoandone.tms.realtime.utils.DateFormatUtil;
import com.twoandone.tms.realtime.utils.KafkaUtil;
import com.twoandone.tms.realtime.utils.MyClickHouseUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.time.Duration;
import java.util.concurrent.TimeUnit;

public class DwsWaybillOrderCargoAll {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        //并行度的设置
        KafkaSource<String> kafkaSource = KafkaUtil.getKafkaSource("dwd_waybill_order_cargo", "dwd_waybill_order_cargo");
        //消费kafka数据,分装为数据流
        DataStreamSource<String> kfDS = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka_source");
        //kfDS.print();
        //对数据进行ETL以及类型的转换
        SingleOutputStreamOperator<JSONObject> jsonDS = kfDS.process(new ProcessFunction<String, JSONObject>() {
            @Override
            public void processElement(String jsonStr, ProcessFunction<String, JSONObject>.Context context, Collector<JSONObject> out) throws Exception {
                JSONObject jsonOj = JSON.parseObject(jsonStr);
                out.collect(jsonOj);
            }
        });
        //jsonDS.print();
        //设置水位线
        SingleOutputStreamOperator<JSONObject> withWaterDS = jsonDS.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                    @Override
                    public long extractTimestamp(JSONObject jsonObject, long l) {
                        return jsonObject.getLong("ts") * 1000;
                    }
                }));
        //withWaterDS.print();
        //根据运单id进行分组
        KeyedStream<JSONObject, String> keyedDS = withWaterDS.keyBy(new KeySelector<JSONObject, String>() {
            @Override
            public String getKey(JSONObject jsonObject) throws Exception {
                return jsonObject.getString("order_id");
            }
        });
        SingleOutputStreamOperator<JSONObject> processDS = keyedDS.process(new KeyedProcessFunction<String, JSONObject, JSONObject>() {
            private ValueState<JSONObject> valueState;

            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<JSONObject> stateDescriptor = new ValueStateDescriptor<>("lastValueState", JSONObject.class);
                stateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.seconds(10)).build());
                valueState = getRuntimeContext().getState(stateDescriptor);
            }

            @Override
            public void processElement(JSONObject jsonObject, Context ctx, Collector<JSONObject> out) throws Exception {
                JSONObject lastvaluestate = valueState.value();
                //出现了数据的重复
                if (lastvaluestate != null) {
                    String amount = jsonObject.getString("amount");
                    lastvaluestate.put("amount", "-" + amount);
                    out.collect(lastvaluestate);
                }
                valueState.update(jsonObject);
                out.collect(jsonObject);
            }
        });
        //将流中的数据转换为实体类对象
        SingleOutputStreamOperator<OrderCityBean> orderCityBeanDS = processDS.map(new MapFunction<JSONObject, OrderCityBean>() {
            @Override
            public OrderCityBean map(JSONObject jsonObject) throws Exception {
                String orderId = jsonObject.getString("order_id");
                String org_id = jsonObject.getString("org_id");
                String senderProvinceId = jsonObject.getString("sender_province_id");
                String senderCityId = jsonObject.getString("sender_city_id");
                String senderDistrictId = jsonObject.getString("sender_district_id");
                String cargoTypeName = jsonObject.getString("cargo_type_name");
                String amount = jsonObject.getString("amount");
                Long ts = jsonObject.getLong("ts") * 1000;
                OrderCityBean orderCityBean = OrderCityBean.builder()
                        .orderId(orderId)
                        .orgId(org_id)
                        .provinceId(senderProvinceId)
                        .cityId(senderCityId)
                        .districtId(senderDistrictId)
                        .cargoType(cargoTypeName)
                        .amount(new BigDecimal(amount))
                        .ts(ts)
                        .build();
                return orderCityBean;
            }
        });
        //orderCityBeanDS.print();
        //指定水位线,提取事件时间字段
        SingleOutputStreamOperator<OrderCityBean> withWatermarkDS = orderCityBeanDS.assignTimestampsAndWatermarks(
                WatermarkStrategy.<OrderCityBean>forBoundedOutOfOrderness(Duration.ofSeconds(10))
                        .withTimestampAssigner(new SerializableTimestampAssigner<OrderCityBean>() {
                            @Override
                            public long extractTimestamp(OrderCityBean orderCityBean, long l) {
                                return orderCityBean.getTs();
                            }
                        })
        );
        //开窗
        AllWindowedStream<OrderCityBean, TimeWindow> windowDS = withWatermarkDS.windowAll(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)));
        //聚合
        SingleOutputStreamOperator<OrderCityBean> reduceDS = windowDS.reduce(new ReduceFunction<OrderCityBean>() {
            @Override
            public OrderCityBean reduce(OrderCityBean t0, OrderCityBean t1) throws Exception {
                t0.setAmount(t0.getAmount().add(t1.getAmount()));
                return t0;
            }
        }, new AllWindowFunction<OrderCityBean, OrderCityBean, TimeWindow>() {
            @Override
            public void apply(TimeWindow timeWindow, Iterable<OrderCityBean> iterable, Collector<OrderCityBean> collector) throws Exception {
                for (OrderCityBean orderCityBean : iterable) {
                    orderCityBean.setStt(DateFormatUtil.toYmdHms(timeWindow.getStart()));
                    orderCityBean.setEdt(DateFormatUtil.toYmdHms(timeWindow.getEnd()));
                    collector.collect(orderCityBean);
                }

            }
        });
        reduceDS.print();
        //关联地区维度
        SingleOutputStreamOperator<OrderCityBean> withProvince = AsyncDataStream.unorderedWait(
                reduceDS,
                new DimAsyncFunction<OrderCityBean>("DIM_BASE_REGION_INFO") {
                    @Override
                    public void join(OrderCityBean obj, JSONObject dimInfoJsonObj) {
                        obj.setProvinceName(dimInfoJsonObj.getString("TM_NAME"));
                    }

                    @Override
                    public String getKey(OrderCityBean obj) {
                        return obj.getProvinceId();
                    }
                }, 60,
                TimeUnit.SECONDS

        );

        reduceDS.addSink(MyClickHouseUtil.getSinkFunction("insert into dws_waybill_order_cargo_window values(?,?,?,?,?,?,?,?,?,?,?,?,?,?)"));
        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
